package com.qust.EnDecryption.zookeeper;

import com.alibaba.fastjson2.JSON;
import com.qust.EnDecryption.common.Constant;
import com.qust.EnDecryption.common.MetaData;
import com.qust.util.NettyUtils;
import io.netty.channel.Channel;
import io.netty.channel.EventLoopGroup;
import lombok.Data;
import lombok.extern.log4j.Log4j2;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

@Data
@Log4j2
public class ZkNodeProvide {
    private CuratorFramework zkClient;
    private String zkNodeName;
    private Channel nextNode;
    private boolean connectNextNodeFlag;
    private List<NeedChannel> needChannels = new ArrayList<>();

    public ZkNodeProvide(CuratorFramework curatorFramework) {
        this.zkClient = curatorFramework;
    }

    public ZkNodeProvide() {
        this.zkClient = connectZk();
    }

    public ZkNodeProvide(NeedChannel needChannel) {
        this();
        this.needChannels.add(needChannel);
    }

    public ZkNodeProvide(CuratorFramework curatorFramework, NeedChannel... needChannel) {
        this(curatorFramework);
        needChannels.addAll(Arrays.asList(needChannel));
    }

    public static CuratorFramework connectZk(RetryPolicy retryPolicy, int connectTimeoutMs, int sessionTimeoutMs) {
        CuratorFramework cf = CuratorFrameworkFactory.builder()
                .connectionTimeoutMs(connectTimeoutMs)
                .connectString(Constant.zookeeperAddress)
                .sessionTimeoutMs(sessionTimeoutMs)
                .retryPolicy(retryPolicy)
                .namespace("EnDecryption")
                .build();
        cf.start();
        return cf;
    }

    public static CuratorFramework connectZk() {
        //1 重试策略：初试时间为1s 重试10次
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 10);
        return connectZk(retryPolicy, 3000, 5000);
    }

    public void register(ZkNode zkNode) {
        try {
            List<String> children = zkClient.getChildren().forPath("/server");
            for (String child : children) {
                String address = child.split(";")[0];
                if (zkNode.getAddress().equals(address)) {
                    zkClient.delete().forPath("/server/" + child);
                }
            }
            String temp = this.zkClient.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath("/server/" + zkNode.getAddress() + ";", JSON.toJSONBytes(zkNode));
            zkNodeName = temp.split("/")[2];

            getMeta();

        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }


    private void getMeta() throws Exception {
        byte[] bytes = zkClient.getData().forPath("/config/meta");
        Constant.META_DATA = JSON.parseObject(bytes).to(MetaData.class);
    }

    /**
     * 连接下一个节点
     *
     * @param eventLoopGroup
     * @param handle
     * @return
     */
    public Channel connectNextNode(EventLoopGroup eventLoopGroup, NettyUtils.MessageHandle handle) {
        try {
            List<String> list = zkClient.getChildren().forPath("/server");
            list.sort((o1, o2) -> {
                int i1 = Integer.parseInt(o1.split(";")[1]);
                int i2 = Integer.parseInt(o2.split(";")[1]);
                return i1 - i2;
            });

            if (list.get(list.size() - 1).equals(zkNodeName)) {
                log.debug("处于队尾,不连接");
                this.nextNode = null;
                return null;
            } else {
                int i = list.indexOf(zkNodeName);
                String ipAndPort = list.get(i + 1).split(";")[0];
                Constant.nextNodeAddress = ipAndPort;
                log.debug("连接 {}", ipAndPort);
                String[] split = ipAndPort.split(":");
                String ip = split[0];
                int port = Integer.parseInt(split[1]);
                this.nextNode = NettyUtils.generateChannel(ip, port, eventLoopGroup, handle);
                connectNextNodeFlag = true;
                return this.nextNode;
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 添加监视,节点变化时用于重新连接
     *
     * @param eventLoopGroup
     * @param handle
     */
    public void setWatch(EventLoopGroup eventLoopGroup, NettyUtils.MessageHandle handle) {
        PathChildrenCache pathChildrenCache = new PathChildrenCache(zkClient, "/server", false);
        pathChildrenCache.getListenable().addListener((curatorFramework, pathChildrenCacheEvent) -> {

            if (pathChildrenCacheEvent.getType().equals(PathChildrenCacheEvent.Type.CHILD_ADDED)) {
                log.debug("新增节点");
                if (!connectNextNodeFlag) {
                    Channel channel = connectNextNode(eventLoopGroup, handle);
                    for (NeedChannel needChannel : needChannels) {
                        needChannel.need(channel);
                    }
                }
            } else if (pathChildrenCacheEvent.getType().equals(PathChildrenCacheEvent.Type.CHILD_REMOVED)) {
                log.debug("节点退出");
                Channel channel = connectNextNode(eventLoopGroup, handle);
                for (NeedChannel needChannel : needChannels) {
                    needChannel.need(channel);
                }
            }
        });
        try {
            pathChildrenCache.start();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    public void addNeedChannel(NeedChannel needChannel) {
        this.needChannels.add(needChannel);
    }

}
